package com.ydl.learning.flink.demo

import java.util.concurrent.TimeUnit

import com.alibaba.fastjson.JSON
import com.ydl.learning.flink.demo.entity.TestEntity
import com.ydl.learning.flink.demo.myFunction.{AsyncFuncion, AsyncFuncion2}
import com.ydl.learning.flink.demo.mySink.LogSink
import org.apache.flink.streaming.api.scala._

/**
 * 异步demo
 *
 * @author ydl
 * @since 2021/4/14
 */
object AsyncStreamDemo extends Utils {
  def main(args: Array[String]): Unit = {
    val info = getSocketTextStream(9999)
    //多个异步算子并行时需要禁用处理链，否则会导致应该并发处理的多个异步算子变成串行处理
    info._1.disableOperatorChaining()
    val value = info._2.map(JSON.parseObject(_, classOf[TestEntity]))
    val value1 = AsyncDataStream.unorderedWait(value, new AsyncFuncion, 20, TimeUnit.SECONDS, 2).name("set name")
    val value2 = AsyncDataStream.unorderedWait(value, new AsyncFuncion2, 20, TimeUnit.SECONDS, 2).name("set age")

    value1.addSink(new LogSink[TestEntity]).name("output name")
    value2.addSink(new LogSink[TestEntity]).name("output age")

    info._1.execute("Socket Window WordCount")
  }
}
